/**
 * 
 */
package com.googlecode.xcarve.zk;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.config.refer.NotifyListener;
import com.googlecode.xcarve.util.ConcurrentHashSet;


/**
 * @author duxinyun
 *
 */
public class ZookeeperClient extends AbstractZkClient{

	private final ZkClient zkClient;
	static final Logger logger = LoggerFactory.getLogger(ZookeeperClient.class);
	protected final Set<String> registered = new ConcurrentHashSet<String>();
	protected final Set<String> retryRegistered = new ConcurrentHashSet<String>();
	protected final Set<String> unRegistered = new ConcurrentHashSet<String>();
	private volatile KeeperState state = KeeperState.SyncConnected;
	
	public ZookeeperClient(String url){
		zkClient = new ZkClient(url);
		zkClient.subscribeStateChanges(new IZkStateListener() {
			
			@Override
			public void handleStateChanged(KeeperState state) throws Exception {
				ZookeeperClient.this.state = state;
				if(state == KeeperState.Disconnected){
					
				}
			}
			
			@Override
			public void handleSessionEstablishmentError(Throwable error)
					throws Exception {
			}
			
			@Override
			public void handleNewSession() throws Exception {
				recreate();
			}
		});
		logger.info("establish zookeeper connection with address url = {}",url);
	}

	protected void createEphemeral(String path) {
		try {
			zkClient.createEphemeral(path);
		} catch (Exception e) {
		}
	}

	@Override
	protected void createPersistent(String path) {
		try {
			zkClient.createPersistent(path);
		} catch (Exception e) {
		}
	}
	
	void doSubScrible(final String path,final NotifyListener listener){
		zkClient.subscribeChildChanges(path, new IZkChildListener() {
			
			@Override
			public void handleChildChange(String parentPath, List<String> currentChilds)
					throws Exception {
					logger.info("service children has chanaged. parentPath={}", path);
					listener.notify(parentPath,currentChilds);
			}
		});
	}
	
	public boolean exist(String path){
		 try {
			 return zkClient.exists(path);
		 } catch (Exception e) {
			 logger.error("use zookeeper error." , e );
			 throw new IllegalStateException("use zookeeper check path is exist error.", e);
		}
	}

	@Override
	public List<String> getChildren(String path) {
		try {
			return zkClient.getChildren(path);
		} catch (Exception e) {
			logger.error("abort zk childPath error.", e);
			throw new IllegalStateException("abort zk childPath error.", e);
		}
	}
	
	public boolean isAvailable(){
		return state == KeeperState.SyncConnected;
	}
	
	protected void recreate() {
		 Set<String> registerUrl = new HashSet<String>(registered);
		 for (String url : registerUrl) {
			  retryRegistered.add(url);
		}
	}
	
	@Override
	public void retry() {
		Set<String> registerUrl = new HashSet<String>(retryRegistered);
		for(String url : registerUrl){
			try {
				createPath(url, true);
				retryRegistered.remove(url);
			} catch (Exception e) {
				logger.error("register service error.", e);
			}
		}
		
	}
	
}
